package com.joizhang.reactor.framework;

import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;

/**
 * @author joizhang
 */
public abstract class AbstractNioChannel {

    private final SelectableChannel channel;
    private final ChannelHandler handler;
    private final Map<SelectableChannel, Queue<Object>> channelToPendingWrites = new ConcurrentHashMap<>();
    private NioReactor reactor;

    public AbstractNioChannel(ChannelHandler handler, SelectableChannel channel) {
        this.handler = handler;
        this.channel = channel;
    }

    void setReactor(NioReactor reactor) {
        this.reactor = reactor;
    }

    public SelectableChannel getJavaChannel() {
        return channel;
    }

    /**
     * channel 感兴趣的操作
     * @return interested operation.
     */
    public abstract int getInterestedOps();

    public abstract void bind() throws IOException;

    /**
     * 读取 SelectionKey 中的数据并返回
     * @param key 读事件的 SelectionKey
     * @return 读取的数据
     * @throws IOException 如果发生 I/O 错误
     */
    public abstract Object read(SelectionKey key) throws IOException;

    public ChannelHandler getHandler() {
        return handler;
    }

    void flush(SelectionKey key) throws IOException {
        Queue<Object> pendingWrites = channelToPendingWrites.get(key.channel());
        while (true) {
            Object pendingWrite = pendingWrites.poll();
            if (pendingWrite == null) {
                reactor.changeOps(key, SelectionKey.OP_READ);
                break;
            }

            doWrite(pendingWrite, key);
        }
    }

    /**
     * 将数据写入 channel
     * @param pendingWrite 即将写入 channel 的数据
     * @param key 可写的 SelectionKey
     * @throws IOException 如果发生 I/O 错误
     */
    protected abstract void doWrite(Object pendingWrite, SelectionKey key) throws IOException;

    public void write(Object data, SelectionKey key) {
        Queue<Object> pendingWrites = this.channelToPendingWrites.get(key.channel());
        if (pendingWrites == null) {
            synchronized (this.channelToPendingWrites) {
                pendingWrites = this.channelToPendingWrites.get(key.channel());
                if (pendingWrites == null) {
                    pendingWrites = new ConcurrentLinkedDeque<>();
                    this.channelToPendingWrites.put(key.channel(), pendingWrites);
                }
            }
        }
        pendingWrites.add(data);
        reactor.changeOps(key, SelectionKey.OP_WRITE);
    }
}
